package com.lxj.app.dim;

// 添加以下导入语句

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lxj.bean.TableProcess;
import com.lxj.utils.MyKafkaUtil;
import com.lxj.utils.MyPhoenixSink;
import com.lxj.utils.MysqlUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class DimApp {

    /**
     * 主方法，负责初始化Flink环境、配置检查点、设置状态后端、读取Kafka数据并进行处理
     * @param args 命令行参数
     * @throws Exception 可能抛出的异常
     */


    public static void main(String[] args) throws Exception {

        // TODO 1. 初始化环境
        // 获取Flink流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置作业并行度为1，确保处理顺序性
        env.setParallelism(1);

        // 配置检查点相关参数
        // 每5分钟触发一次检查点，保证精确一次语义
        env.enableCheckpointing(5*60000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置检查点超时时间为10分钟
        env.getCheckpointConfig().setCheckpointTimeout(10*60000L);

        // TODO 2. 启用状态后端
        // 使用HashMap作为状态后端，用于存储作业状态
        env.setStateBackend(new HashMapStateBackend());
        // 设置检查点存储路径为HDFS，保证状态持久化
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/2212a/ck");
        // 设置Hadoop操作用户名为ROOT，用于HDFS权限控制
        System.setProperty("HADOOP_USER_NAME", "ROOT");

        // TODO 3. 读取数据 读取kafka主题，topic_db
        // 定义Kafka主题和消费者组ID
        String topic  = "topic_db";
        String groupId = "dimapp";

        // 从Kafka中读取数据，创建数据流
        DataStreamSource<String> ds = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        // 对原始数据流进行过滤处理，只保留特定操作类型的数据
        SingleOutputStreamOperator<JSONObject> filterJSONDS = ds.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    // 将字符串解析为JSON对象
                    JSONObject jsonObject = JSON.parseObject(s);
                    // 获取操作类型字段
                    String type = jsonObject.getString("type");
                    // 只保留insert、update和bootstrap-insert类型的数据
                    if (type.equals("insert") || type.equals("update") || type.equals("bootstrap-insert")) {
                        collector.collect(jsonObject);
                    }
                } catch (Exception ex) {
                    // 捕获并打印格式错误的数据
                    System.out.println("格式错误" + s);
                }
            }
        });

        // 打印过滤后的数据，用于调试
        filterJSONDS.print("过滤后数据>>>>");

        // 使用FlinkCDC读取mysql配置信息表，创建配置流
        // 纬度表的配置信息存储在mysql的table_process表中
        DataStream<String> mysqlDs = MysqlUtil.cdcMysql(env, "gmall_config", "table_process");

        // 创建Map状态描述符，用于存储表处理配置
        MapStateDescriptor<String, TableProcess> mapState =
                new MapStateDescriptor<>("mapState", String.class, TableProcess.class);

        // 将配置流广播，使所有并行任务都能访问配置信息
        BroadcastStream<String> broadcastStream = mysqlDs.broadcast(mapState);

        // 连接主流(业务数据)和广播流(配置信息)
        BroadcastConnectedStream<JSONObject, String> connect = filterJSONDS.connect(broadcastStream);

        // 处理连接后的流：
        // 广播流包含纬度表配置信息
        // 主流包含业务数据，需要根据配置信息进行路由
        // 匹配成功的业务数据将被写入对应的纬度表
        SingleOutputStreamOperator<JSONObject> dimDs = connect.process(new TableProcessFunction(mapState));

        // 打印最终处理后的数据，用于调试
        dimDs.print("最终数据>");

        // 将处理后的数据写入Phoenix(HBase)
        dimDs.addSink(new MyPhoenixSink());

        // 执行Flink作业
        env.execute();
    }








}
